package com.hzgj.bcl.soa.cicada.handle;

import com.hzgj.bcl.soa.cicada.pool.Channels;
import com.hzgj.bcl.soa.cicada.sender.CicadaSender;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.Util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Objects;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class NettyHandler extends SimpleChannelInboundHandler {

    private final Processor protocol;
    private final CicadaSender cicadaSender;
    private Logger logger = LoggerFactory.getLogger(getClass());

    public NettyHandler(Processor protocol) {
        this.protocol = protocol;
        Objects.requireNonNull(protocol.getCicadaSender());
        this.cicadaSender = protocol.getCicadaSender();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {

        Channel channel = ctx.channel();
        Object attribute = Channels.getAttribute(channel);
        if (attribute instanceof ResponseFuture) {
            ResponseFuture<Object> future = (ResponseFuture<Object>) attribute;

//            if (((Message) msg).getHeader(Constants.MESSAGE_SID) == null) {
//                ((Message) msg).putHeader(Constants.MESSAGE_SID, future.getMessageId());
//            }
            protocol.handle(channel, future, msg);
        } else if (attribute instanceof Map) {
            Map<?, ?> map = (Map<?, ?>) attribute;
            ResponseFuture<Object> future = (ResponseFuture<Object>) map.remove(((Message) msg).getHeader(Constants.MESSAGE_SID));
            if (future != null) {
                protocol.handle(channel, future, msg);
            }
        } else if (attribute != Channels.DiscardEvent.INSTANCE) {
            // unhandled send_message
            logger.debug("Orphan channel {} with attribute {} received send_message {}, closing", channel, attribute, msg);
            releaseChannel(channel);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        Channel channel = ctx.channel();
        try {
            super.channelInactive(ctx);
        } catch (Exception ex) {
            logger.trace("super.channelClosed", ex);
        }

        Object attribute = Channels.getAttribute(channel);
        logger.info("Channel Closed: {} with attribute {}", channel, attribute);

        if (attribute == null) {
            logger.error("channel " + channel + " can't contain any attribute.");
            releaseChannel(channel);
        }

        if (attribute instanceof ResponseFuture<?>) {
            inactive(ResponseFuture.class.cast(attribute), channel);
        } else if (attribute instanceof Map) {
            ((Map<Integer, ResponseFuture<Object>>) attribute).values().stream().forEach(future -> inactive(future, channel));
        }
    }

    private void inactive(ResponseFuture<Object> future, Channel channel) {
        if (!cicadaSender.isOk()) {
            return;
        }
        future.touch();
        protocol.onClose(future);
        cicadaSender.handleUnexpectedClosedChannel(channel, future, Util.REMOTELY_CLOSED_EXCEPTION);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
        Throwable cause = e.getCause() != null ? e.getCause() : e;
        Channel channel = ctx.channel();
        logger.info("Unexpected I/O exception on channel {}", channel, cause);
        Object attribute = Channels.getAttribute(channel);

        if (attribute == null) {
            logger.error("channel " + channel + " can't contain any attribute.");
            releaseChannel(channel);
        }
        if (attribute instanceof ResponseFuture<?>) {
            exception(channel, ResponseFuture.class.cast(attribute), e, cause);
        } else if (attribute instanceof Map) {
            ((Map<Integer, ResponseFuture<Object>>) attribute).values().stream().forEach(future -> exception(channel, future, e, cause));
        }
    }

    private void exception(Channel channel, ResponseFuture<Object> future, Throwable e, Throwable cause) {
        if (!cicadaSender.isOk()) {
            return;
        }
        if (future != null) {
            try {
                future.detachChannel();
                cicadaSender.abortAndReleaseChannel(channel, future, cause);
                protocol.onError(future, e);
            } catch (Throwable t) {
                logger.error(t.getMessage(), t);
            }
        } else {
            releaseChannel(channel, future);
        }
    }

    private void releaseChannel(Channel channel, ResponseFuture<Object> future) {
        cicadaSender.releaseChannel(channel, future.getPoolKey());
    }

    private void releaseChannel(Channel channel) {
        cicadaSender.releaseChannel(channel);
    }
}